package com.xiaoming.zk;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.io.File;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.util.*;


public class CuratorServer {
    private static  CuratorFramework client = null;
    private static Properties properties = new Properties();
    private static CuratorServer curatorServer = null;

    private Map<Class,String> nodeMap = new HashMap<>();

    /***
     * 开始构建
     */
    public void build(Class clazz, ZnodeDataChangeHandler dataChangeHandler) throws Exception {
        String packageName = clazz.getName();
        String packagePath = packageName.substring(0,packageName.lastIndexOf( "."));
        // 类加载
        loadClass(packagePath);
        // 创建节点
        createNode();
        // 注册监听
        registerWatcher(dataChangeHandler);
    }

    /***
     * 设置数据
     * @param obj
     * @throws Exception
     */
    public void setData(Object obj) throws Exception {
        Class<?> aClass = obj.getClass();
        String nodeName = nodeMap.get(aClass);
        if(nodeName == null){
            throw new RuntimeException("当前对象不存在");
        }
        Field[] fields = aClass.getDeclaredFields();
        Map<String ,Object> map = new HashMap<>();
        for (Field field:fields) {
            if(field.isAnnotationPresent(PropertyName.class)){
                PropertyName property = field.getAnnotation(PropertyName.class);
                field.setAccessible(true);
                map.put(property.value(),field.get(obj));
            }
        }
        client.setData().forPath(nodeName, JSON.toJSONBytes(map));
    }

    /***
     * 创建节点
     * @throws Exception
     */
    private void createNode() throws Exception {
        Collection<String> values = nodeMap.values();
        for (String nodeName : values){
            Stat stat = client.checkExists().forPath(nodeName);
            if(stat == null){
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
                        .forPath(nodeName);
            }
        }
    }
    /***
     * 读取配置存入到 Spring 容器
     * @param packageName
     */
    private void loadClass(String packageName) throws Exception {
            String packagePath = Thread.currentThread().getContextClassLoader()
                    .getResource("").getPath()
                    + packageName.replaceAll("\\.", "/");
            File pack = new File(packagePath);
            File[] files = pack.listFiles();
            for (File file : files) {
                if (file.isDirectory()) {
                    loadClass(packageName + "." + file.getName());
                } else if (file.getName().endsWith(".class")) {
                    String fileName = file.getName();
                    String className = packageName + "." + fileName.substring(0, fileName.lastIndexOf(".class"));
                    Class<?> aClass = Class.forName(className);
                    if(aClass.isAnnotationPresent(Znode.class)){
                        Znode znode = aClass.getAnnotation(Znode.class);
                        nodeMap.put(aClass, znode.value());
                    }
                }
            }
    }

    /***
     * 注册监听
     * @throws Exception
     * @param dataChangeHandler
     */
    private void registerWatcher(ZnodeDataChangeHandler dataChangeHandler) throws Exception {
        Set<Map.Entry<Class, String>> entries = nodeMap.entrySet();
        for (Map.Entry<Class, String> entry : entries){
            Class aClass = entry.getKey();
            String nodeName = entry.getValue();
            Object instance = aClass.getDeclaredConstructor().newInstance();
            Watcher watcher = watchedEvent -> {
                if (watchedEvent.getType() == Watcher.Event.EventType.NodeDataChanged) {
                    try {
                        byte[] bytes = client.getData().forPath(nodeName);
                        JSONObject json = JSONObject.parseObject(new String(bytes));
                        System.out.println(String.format("节点 %s 数据改变为：%s",nodeName,json));
                        Field[] fields = aClass.getDeclaredFields();
                        for (Field field : fields) {
                            if (field.isAnnotationPresent(PropertyName.class)) {
                                String propertyName = field.getAnnotation(PropertyName.class).value();
                                field.setAccessible(true);
                                field.set(instance, json.get(propertyName));
                            }
                        }
                        dataChangeHandler.change(instance);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                try {
                    registerWatcher(dataChangeHandler);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            };
            client.getData().usingWatcher(watcher).forPath(nodeName);
        }
    }

    /***
     * 初始化全局鼠标
     * @throws Exception
     */
    private void init() throws Exception {
        InputStream in = CuratorServer.class.getClassLoader().getResourceAsStream("zookeeper.properties");
        CuratorServer.properties.load(in);
        String namespace = CuratorServer.properties.getProperty("zookeeper.namespace");
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
        String zkAddress = CuratorServer.properties.getProperty("zookeeper.server.address");
        CuratorServer.client = CuratorFrameworkFactory.builder()
                .connectString(zkAddress)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .namespace(namespace)
                .build();
        CuratorServer.client.start();
    }
    public static CuratorServer getInstance() throws Exception {
        if(curatorServer == null){
            curatorServer = new CuratorServer();
            curatorServer.init();
        }
        return curatorServer;
    }
    private CuratorServer(){
    }
}
